package com.baihejia.common.dubbo.server;

import com.baihejia.common.dubbo.domain.Page;
import com.baihejia.common.dubbo.libs.JsonHelper;
import com.baihejia.common.dubbo.service.Api;
import com.google.common.collect.Lists;

import org.apache.commons.collections.CollectionUtils;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequestBuilder;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.util.*;
import java.util.stream.Collectors;

public abstract class Es<T, Q> {

    private String indexAlias;

    private String type;

    protected static Api api = new ApiImpl();

    public Es(String indexAlias, String type) {
        this.indexAlias = indexAlias;
        this.type = type;
    }

    protected static Client client;

    static {
        try {
            System.setProperty("es.set.netty.runtime.available.processors", "false");
            Settings settings = Settings.builder().put("cluster.name", Configuration.get("es.cluster_name")).build();
            String esUrls = Configuration.get("es.url");
            String[] esUrlArray = esUrls.split(";|,");

            List<InetSocketTransportAddress> inetSocketTransportAddressList = new ArrayList<>();
            for (String esUrl : esUrlArray) {
                String[] esConfArray = esUrl.split(":");
                String esHost = esConfArray[0];
                String esPort = esConfArray[1];
                inetSocketTransportAddressList.add(new InetSocketTransportAddress(InetAddress.getByName(esHost), Integer.valueOf(esPort)));
            }

            client = new PreBuiltTransportClient(settings).addTransportAddresses(inetSocketTransportAddressList.toArray(new InetSocketTransportAddress[inetSocketTransportAddressList.size()]));

            Runtime.getRuntime().addShutdownHook(new Thread() {
                public void run() {
                    release();
                }
            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    protected static final int SEARCH_LIMIT = 10000;

    protected static final DistanceUnit DISTANCE_UNIT = DistanceUnit.KILOMETERS;

    protected void setIndexAlias(String indexAlias) {
        this.indexAlias = indexAlias;
    }

    protected String getIndexAlias() {
        return this.indexAlias;
    }

    protected void setType(String type) {
        this.type = type;
    }

    protected String getType() {
        return this.type;
    }

    protected abstract XContentBuilder buildMapping() throws Exception;

    protected abstract void rebuildIndex(String index);

    public boolean rebuildIndex() throws Exception {
        String index = String.format("%s-v-%s", indexAlias, System.currentTimeMillis());
        client.admin().indices().prepareCreate(index).execute().actionGet();
        XContentBuilder contentBuilder = buildMapping();
        PutMappingResponse response = client
                .admin()
                .indices()
                .preparePutMapping(index)
                .setType(type)
                .setSource(contentBuilder)
                .execute()
                .actionGet();
        rebuildIndex(index);
        swapIndex(index);
        return response.isAcknowledged();
    }

    private void swapIndex(String index) {
        List<String> oldIndexes = Lists.newArrayList();

        //find all old indexes
        if (client.admin().indices().prepareAliasesExist(indexAlias).execute().actionGet().exists()) {
            final Iterator<String> iterator = client.admin().indices().prepareGetAliases(indexAlias).execute().actionGet().getAliases().keysIt();

            while (iterator.hasNext()) {
                oldIndexes.add(iterator.next());
            }
        }
        // swap alias
        final IndicesAliasesRequestBuilder indicesAliasesRequestBuilder = client.admin().indices().prepareAliases();
        if (!oldIndexes.isEmpty()) {
            indicesAliasesRequestBuilder.removeAlias(oldIndexes.toArray(new String[oldIndexes.size()]), indexAlias);
        }
        indicesAliasesRequestBuilder.addAlias(index, indexAlias);
        indicesAliasesRequestBuilder.execute().actionGet();

        // remove old indexes
        if (!oldIndexes.isEmpty()) {
            client.admin().indices().prepareDelete(oldIndexes.toArray(new String[oldIndexes.size()])).execute().actionGet();
        }

        System.out.println("old:" + oldIndexes);
        System.out.println("new:" + index);

    }

    public static void release() {
        client.close();
    }

    private List<String> findRealIndex() {

        List<String> realIndexes = Lists.newArrayList();
        //find all real indexes
        if (client.admin().indices().prepareAliasesExist(indexAlias).execute().actionGet().exists()) {
            final Iterator<String> iterator = client.admin().indices().prepareGetAliases(indexAlias).execute().actionGet().getAliases().keysIt();

            while (iterator.hasNext()) {
                realIndexes.add(iterator.next());
            }
        }

        return realIndexes;
    }

    public void build(Map<String, Object> map) throws Exception{
        client.prepareIndex(findRealIndex().get(0), type, String.valueOf(map.get("id")))
                .setSource(map, XContentType.JSON)
                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
                .execute()
                .actionGet();
    }

    public void update(Map<String, Object> data) throws Exception{
        client.prepareUpdate(findRealIndex().get(0), type, String.valueOf(data.get("id")))
                .setDoc(JsonHelper.toString(data), XContentType.JSON)
                .setRefreshPolicy(RefreshPolicy.IMMEDIATE)
                .execute()
                .actionGet();
    }
    
    public void deleteItem(String id) {
        client.prepareDelete(findRealIndex().get(0), type, id)
                .execute()
                .actionGet();
    }

    public void bulkBuild(List<T> entities, String index) throws Exception {

        if (CollectionUtils.isEmpty(entities)) {
            return;
        }

        BulkRequestBuilder prepareBulk = client.prepareBulk();
        for (T entity : entities) {
            Map map = toMap(entity);
            Object idObject = null;
            try {
                idObject = map.get("id");
            } catch (Exception ignored) {
            }
            if (idObject != null) {
                prepareBulk.add(client.prepareIndex(index, type, String.valueOf(map.get("id"))).setSource(map, XContentType.JSON));
            } else {
                prepareBulk.add(client.prepareIndex(index, type).setSource(map, XContentType.JSON));
            }
        }
        prepareBulk.execute().actionGet();
    }

    protected abstract void putExtraFields(Map<String, Object> jsonObject, T entity);

    private Map<String, Object> toMap(T entity) throws Exception {

        Map<String, Object> map = JsonHelper.toMap(entity);

        putExtraFields(map, entity);

        return map;
    }

    protected abstract SearchHits esHits(Q query) throws Exception;

    public Page<String> search(Q query) throws Exception {
        SearchHits searchHits = esHits(query);
        long showCount = searchHits.getTotalHits();
        long count = showCount > SEARCH_LIMIT ? SEARCH_LIMIT : showCount;
        return new Page<>(Arrays.asList(searchHits.getHits()).stream().map(SearchHit::getId).collect(Collectors.toList()), count, showCount);
    }
}
